package com.study.flink.demo;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import com.study.flink.entity.does.HumidityDO;
import com.study.flink.entity.does.TemperatureDO;
import com.study.flink.source.HumiditySource;
import com.study.flink.source.TemperatureSource;

import lombok.extern.slf4j.Slf4j;

/**
 * 测试Flink里的join操作。
 * 
 * @projectName: flink-demo
 * @author: yandong
 * @description: 数据源：蜂巢的传感数据。
 * @date: 2019年7月21日 下午9:25:54
 * @version:
 */
@Slf4j
public class HumidityJoinTemperatureDemo {
    public void run(String[] args) {
        intervalJoin(args);
    }

    /**
     * Join操作
     * 
     * @author yandong
     * @description
     * @date 2019年7月22日 下午8:39:52
     * @version
     * @param args
     */
    public void windowJoinFromHumidity(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<HumidityDO> humidityDataStream = env.addSource(new HumiditySource());
        DataStream<TemperatureDO> temperatureDataStream = env.addSource(new TemperatureSource());

        humidityDataStream.join(temperatureDataStream).where(new KeySelector<HumidityDO, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public String getKey(HumidityDO value) throws Exception {
                return value.getTimestamp();
            }
        }).equalTo(new KeySelector<TemperatureDO, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public String getKey(TemperatureDO value) throws Exception {
                return value.getTimestamp();
            }
        }).window(TumblingEventTimeWindows.of(Time.days(1)))
            .apply(new JoinFunction<HumidityDO, TemperatureDO, Tuple3<String, String, String>>() {
                private static final long serialVersionUID = 1L;

                @Override
                public Tuple3<String, String, String> join(HumidityDO first, TemperatureDO second)
                    throws Exception {
                    return new Tuple3<String, String, String>(first.getTimestamp(),
                        null == first ? null : first.getHumidity(),
                        null == second ? null : second.getTemperature());
                }
            }).writeAsText("/data/bigdata/forflink/humidity_join_temperature."
            + new SimpleDateFormat("YYMMddHHmmss").format(new Date()));

        try {
            env.execute();
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * Join操作
     * 
     * @author yandong
     * @description 该例证明：join操作默认是inner join的。
     * @date 2019年7月22日 下午8:39:52
     * @version
     * @param args
     */
    public void windowJoinFromTemperature(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<HumidityDO> humidityDataStream = env.addSource(new HumiditySource());
        DataStream<TemperatureDO> temperatureDataStream = env.addSource(new TemperatureSource());

        temperatureDataStream.join(humidityDataStream)
            .where(new KeySelector<TemperatureDO, String>() {
                private static final long serialVersionUID = 1L;

                @Override
                public String getKey(TemperatureDO value) throws Exception {
                    return value.getTimestamp();
                }
            }).equalTo(new KeySelector<HumidityDO, String>() {
                private static final long serialVersionUID = 1L;

                @Override
                public String getKey(HumidityDO value) throws Exception {
                    return value.getTimestamp();
                }
            }).window(TumblingEventTimeWindows.of(Time.days(1)))
            .apply(new JoinFunction<TemperatureDO, HumidityDO, Tuple3<String, String, String>>() {
                private static final long serialVersionUID = 1L;

                @Override
                public Tuple3<String, String, String> join(TemperatureDO first, HumidityDO second)
                    throws Exception {
                    if (null == second) {
                        log.info("遇见second为空： " + first.getTimestamp());
                    }

                    return new Tuple3<String, String, String>(first.getTimestamp(),
                        null == first ? "" : first.getTemperature(),
                        null == second ? "" : second.getHumidity());
                }
            }).writeAsText("/data/bigdata/forflink/humidity_join_temperature."
                + new SimpleDateFormat("YYMMddHHmmss").format(new Date()));

        try {
            env.execute();
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 对一定间隔的数据join。
     * 
     * @author yandong
     * @description 该例证明也是 inner join。同时不知为什么返回结果比前面两个例子少很多。
     * @date 2019年7月22日 下午9:07:52
     * @version
     * @param args
     */
    public void intervalJoin(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<HumidityDO> humidityStream = env.addSource(new HumiditySource());
        DataStream<TemperatureDO> temperatureStream = env.addSource(new TemperatureSource());

        temperatureStream.keyBy("timestamp").intervalJoin(humidityStream.keyBy("timestamp"))
            .between(Time.hours(-1), Time.hours(1)).upperBoundExclusive().lowerBoundExclusive()
            .process(
                new ProcessJoinFunction<TemperatureDO, HumidityDO, Tuple3<String, String, String>>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public void processElement(TemperatureDO left, HumidityDO right,
                        ProcessJoinFunction<TemperatureDO, HumidityDO, Tuple3<String, String, String>>.Context ctx,
                        Collector<Tuple3<String, String, String>> out) throws Exception {
                        Tuple3<String, String, String> tuple = new Tuple3<String, String, String>();
                        tuple.f0 = null == left ? right.getTimestamp() : left.getTimestamp();
                        tuple.f1 = null == left ? "" : left.getTemperature();
                        tuple.f2 = null == right ? "" : right.getHumidity();
                        out.collect(tuple);
                    }
                })
            .writeAsText("/data/bigdata/forflink/humidity_join_temperature."
                + new SimpleDateFormat("YYMMddHHmmss").format(new Date()));

        try {
            env.execute();
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }
}
